package com.flink.sink.redis;

import com.flink.readsource.ReadDataSourceDemo;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.redis.RedisSink;
import org.apache.flink.streaming.connectors.redis.common.config.FlinkJedisPoolConfig;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisCommand;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisCommandDescription;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisMapper;

/**
 * 描述:
 * 将数据写入redis
 *
 * @author yanzhengwu
 * @create 2022-07-23 22:01
 */
public class SinkToRedis {

    public static void main(String[] args) throws Exception {
        //声明执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //测试为了保证元素顺序并行度设置为4，则每个线程生成一个文件
        env.setParallelism(4);

        DataStreamSource<Integer> integerData = env.addSource(new ReadDataSourceDemo.UserDefinedSource());

        FlinkJedisPoolConfig redisConfig = new FlinkJedisPoolConfig.Builder()
                .setHost("192.168.43.8")
                .build();

        RedisSink<Integer> redisSink = new RedisSink<>(redisConfig, new MyRedisMapper());

        //由于官方没有提供redis的api 所以无法使用flink 1.15 版本的sinkTo方法,只能使用老版本的addSink
        integerData.addSink(redisSink);

        env.execute();

    }

    public static class MyRedisMapper implements RedisMapper<Integer>{

        /**
         * 此方法是指定 redis 做什么命令操作，此处写入的规则是hash的方式来存储的kv形式
         * @return
         */
        @Override
        public RedisCommandDescription getCommandDescription() {
            return new RedisCommandDescription(RedisCommand.HSET);
        }

        /**
         * 以什么规则写入key
         * @param integer
         * @return
         */
        @Override
        public String getKeyFromData(Integer integer) {
            return "number";
        }

        /**
         * 以什么我规则写入value
         * @param integer
         * @return
         */
        @Override
        public String getValueFromData(Integer integer) {
            return String.valueOf(integer);
        }
    }
}
